package cn.devil.two.reportFormStatistics

import cn.devil.two.utils.{CheckArguments, SCUtiles}
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import scalikejdbc.config.DBs
import scalikejdbc.{DB, SQL}
object BusinessLogic {
  def main(args: Array[String]): Unit = {
    CheckArguments(args,1,"请输入读取数据的文件",this.getClass)
    val Array(inputFilePath) = args
    val sc: SparkContext = SCUtiles.createSparkContextInstance("业务逻辑")
    val sqlContext = SCUtiles.createSQLContextInstance(sc)
    val frame: DataFrame = sqlContext.read.parquet(inputFilePath)
    val resultDta = frame.map(t => {
      //获得省
      val province = t.getAs[String]("provincename")
      //获得市
      val city = t.getAs[String]("cityname")
      val requestmode = t.getAs[Int]("requestmode")
      val processnode = t.getAs[Int]("processnode")
      val iseffective = t.getAs[Int]("iseffective")
      val isbilling = t.getAs[Int]("isbilling")
      val isbid = t.getAs[Int]("isbid")
      val adorderid = t.getAs[Int]("adorderid")
      val iswin = t.getAs[Int]("iswin")
      ((province, city), (requestmode, processnode,iseffective,isbilling,isbid,adorderid,iswin))
    })
    resultDta.mapValues(mv => {
      // 原始请求  requestmode = 1  processnode >=1
      val primordial = if (mv._1 == 1 && mv._2 >= 1) 1 else 0
      //有效请求 requestmode = 1  processnode >=2
      val valid = if(mv._1 == 1 && mv._2 >= 2) 1 else 0
      //广告日志请求   requestmode = 1  processnode =3
      val advertising = if(mv._1 == 1 && mv._2 == 3) 1 else 0
      //参与竞价数  iseffective = 1   isbilling=1   isbid=1   adorderid！=0
      val participation = if(mv._3 == 1 && mv._4 == 1 && mv._5 == 1 && mv._6 !=0) 1 else 0
      //竞价成功数 iseffective = 1   isbilling=1     iswin = 1
      val success = if(mv._3 == 1 && mv._4 == 1 && mv._7 == 1) 1 else 0
      //展示数     requestmode = 2     iseffective = 1
      val show = if(mv._1 == 2 && mv._3 == 1) 1 else 0
      //点击数     requestmode = 3   iseffective = 1
      val click = if(mv._1 == 3 && mv._3 >= 1) 1 else 0
      //DSP广告消费 iseffective = 1   isbilling=1     iswin = 1
      val DSPconsumption = if(mv._3 == 1 && mv._4 == 1 && mv._7 == 1) 1 else 0
      //DSP广告成本 iseffective = 1   isbilling=1     iswin = 1
      val DSPcost = if(mv._3 == 1 && mv._4 == 1 && mv._7 == 1) 1 else 0
      (primordial,valid,advertising,participation,success,show,click,DSPconsumption,DSPcost)
    }).reduceByKey((r1,r2)=>{
      val primordial = r1._1 + r2._1
      val valid = r1._2 + r2._2
      val advertising = r1._3 + r2._3
      val participation = r1._4 + r2._4
      val success = r1._5 + r2._5
      val show = r1._6 + r2._6
      val click = r1._7 + r2._7
      val DSPconsumption = r1._8 + r2._8
      val DSPcost = r1._9 + r2._9
      (primordial,valid,advertising,participation,success,show,click,DSPconsumption,DSPcost)
    }).foreachPartition(it =>{



    /*  // 加载配置信息Q
      DBs.setup()
      //开启事务，批量插入
      DB.localTx{implicit session =>
        val seq = it.map(t => {
          Seq(t._1._1, t._1._2, t._2._1,t._2._2,t._2._3,t._2._4,t._2._5,t._2._6,t._2._7,t._2._8,t._2._9)
        }).toSeq
        SQL("insert into leopard_last values(?,?,?,?,?,?,?,?,?,?,?)").batch(seq:_*).apply()
      }*/
    })
    sc.stop()
  }
}
